package com.hzy.kafka.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaConfig {

    @Resource
    private KafkaProperties kafkaProperties;

    /**
     * 定制 接收 配置
     * @return
     */
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
//        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
//        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    /**
     * kafka监听工厂
     * 不自动启动
     * @param configurer
     * @return
     */
    @Bean("batchFactory")
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        // 可以定制 消息队列 接收 的配置
        factory.setConsumerFactory(consumerFactory);
        //开启批量消费功能
        factory.setBatchListener(true);
        //不自动启动
        factory.setAutoStartup(false);
        factory.getContainerProperties().setPollTimeout(1500);
        //配置手动提交offset
        // MANUAL   当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

        // COUNT　当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交，配合 使用
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
//        factory.getContainerProperties().setAckCount(5);

        // TIME     当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于PollTimeout时提交
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.TIME);

        // COUNT_TIME   TIME |　COUNT　有一个条件满足时提交
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.COUNT_TIME);

        // BATCH    当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);

        // RECORD   当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD);

        // MANUAL_IMMEDIATE 手动调用Acknowledgment.acknowledge()后立即提交
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

        configurer.configure(factory, consumerFactory);
        return factory;
    }

}
